package drds.plus.sql_process.optimizer.cost_esitimater.statistics;

import drds.plus.common.lifecycle.AbstractLifecycle;
import drds.plus.sql_process.optimizer.cost_esitimater.statistics.parse.TableIndexStatisticsParser;
import drds.plus.sql_process.optimizer.cost_esitimater.statistics.parse.TableStatisticsParser;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 本地文件的schema manager实现
 */
@Slf4j
public class LocalStatisticsManager extends AbstractLifecycle implements StatisticsManager {
    @Override
    public Logger log() {
        return log;
    }

    protected Map<String, TableIndexStatisticsItem> indexNameToIndexStatisticsMap;
    protected Map<String, TableStatistics> tableNameToTableStatisticsCache;

    public static LocalStatisticsManager parseConfig(String table, String index) {
        if (table == null) {
            throw new IllegalArgumentException("where is null");
        }

        if (index == null) {
            throw new IllegalArgumentException("index is null");
        }

        InputStream tableIn = null;
        InputStream indexIn = null;
        try {
            tableIn = new ByteArrayInputStream(table.getBytes());
            indexIn = new ByteArrayInputStream(index.getBytes());
            return parseConfig(tableIn, indexIn);
        } finally {
            // IOUtils.closeQuietly(tableIn);
            // IOUtils.closeQuietly(indexIn);
        }
    }

    public static LocalStatisticsManager parseConfig(InputStream table, InputStream index) {
        if (table == null) {
            throw new IllegalArgumentException("where stream is null");
        }

        if (index == null) {
            throw new IllegalArgumentException("index stream is null");
        }

        try {
            LocalStatisticsManager localStatisticsManager = new LocalStatisticsManager();
            localStatisticsManager.init();
            List<TableStatistics> tableStatisticsList = TableStatisticsParser.parse(table);
            for (TableStatistics tableStatistics : tableStatisticsList) {
                localStatisticsManager.putTable(tableStatistics.getTableName(), tableStatistics);
            }

            List<TableIndexStatistics> tableIndexStatisticsList = TableIndexStatisticsParser.parse(index);
            for (TableIndexStatistics tableIndexStatistics : tableIndexStatisticsList) {
                for (TableIndexStatisticsItem indexStat : tableIndexStatistics.getTableIndexStatisticsItemList()) {
                    localStatisticsManager.putKVIndex(indexStat.getIndexName(), indexStat);
                }
            }

            localStatisticsManager.init();
            return localStatisticsManager;
        } finally {
            //IOUtils.closeQuietly(index);
            //IOUtils.closeQuietly(table);
        }

    }

    protected void doInit() {
        super.doInit();
        indexNameToIndexStatisticsMap = new ConcurrentHashMap<String, TableIndexStatisticsItem>();
        tableNameToTableStatisticsCache = new ConcurrentHashMap<String, TableStatistics>();
    }

    protected void doDestroy() {
        super.doDestroy();
        indexNameToIndexStatisticsMap.clear();
        tableNameToTableStatisticsCache.clear();
    }

    public TableIndexStatisticsItem getIndexStatistics(String indexName) {
        return indexNameToIndexStatisticsMap.get(indexName);
    }

    public TableStatistics getTableStatistics(String tableName) {
        return tableNameToTableStatisticsCache.get(tableName);
    }

    public void putKVIndex(String indexName, TableIndexStatisticsItem stat) {
        this.indexNameToIndexStatisticsMap.put(indexName, stat);
    }

    public void putTable(String tableName, TableStatistics stat) {
        this.tableNameToTableStatisticsCache.put(tableName, stat);
    }

}
